package org.apache.ocean.transaction;

import java.util.ArrayList;
import java.util.List;

import org.apache.lucene.index.SegmentReader;
import org.apache.lucene.search.HitCollector;
import org.apache.lucene.search.Query;
import org.apache.ocean.CHitQueue;
import org.apache.ocean.CScoreDoc;
import org.apache.ocean.CScoreDocument;
import org.apache.ocean.ClusterUtil;
import org.apache.ocean.SegmentIndexSearcher;
import org.apache.ocean.transaction.Index.IndexSnapshot;
import org.apache.ocean.transaction.Index.IndexSnapshot.SegmentSnapshot;
import org.apache.ocean.util.DocSet;

public class MultiSnapshotSearch {
	IndexSnapshot indexSnapshot;
	
	public MultiSnapshotSearch(IndexSnapshot indexSnapshot) {
		this.indexSnapshot = indexSnapshot;
	}
	
	public List<CScoreDoc> search(Query query, int start, int length) throws Exception {
		int numDocsReturned = start + length;
		CHitQueue queue = new CHitQueue((int) numDocsReturned);
		for (SegmentSnapshot segmentSnapshot : indexSnapshot.segmentSnapshotMap.values()) {
			SegmentHitCollector segmentHitCollector = new SegmentHitCollector(segmentSnapshot.filter, queue, segmentSnapshot.segmentReader);
			SegmentIndexSearcher segmentIndexSearcher = new SegmentIndexSearcher(segmentSnapshot.segmentReader);
			segmentIndexSearcher.search(query, segmentHitCollector);
		}
		List<CScoreDoc> scoreDocs = ClusterUtil.toScoreDocs(start, length, queue);
		return scoreDocs;
	}
	
	public class SegmentHitCollector extends HitCollector {
		DocSet filter;
		CHitQueue queue;
		SegmentReader segmentReader;
		
		public SegmentHitCollector(DocSet filter, CHitQueue queue, SegmentReader segmentReader) {
			this.filter = filter;
			this.queue = queue;
			this.segmentReader = segmentReader;
		}
		
		public void collect(int doc, float score) {
			if (filter != null && !filter.exists(doc)) return;
			CScoreDoc scoreDoc = new CScoreDoc(doc, score, null, segmentReader);
			queue.insert(scoreDoc);
		}
	}
}
